#include "global.h"
#include "globalExtern.h"
#include "cmd.h"
#include "MarkupSTL.h"
#include "Game.h"
#include "Host.h"
#include "Logic.h"
#include "9300_gameinfo_storage.pb.h"

#include <iterator>
#include <unistd.h>


// 交换机类型 -- 精准路由
const char* CGameServer::m_mqExchangeType 			= "direct";
// 交换机名称
const char* CGameServer::m_mqExchange_NonBusiness   = "nonbusiness";
const char* CGameServer::m_mqExchange_Business 		= "business";
// 路由键
const char* CGameServer::m_mqRoutingKey_NonBusiness	= "nonbusiness";
const char* CGameServer::m_mqRoutingKey_Business	= "business";
// 消息队列
const char* CGameServer::m_mqQueue_NonBusiness		= "nonbusiness";
const char* CGameServer::m_mqQueue_Business			= "business";


CGameServer::CGameServer(): m_mqPort(0), m_mqConnection(nullptr), m_mqChannel(1)
{

}

CGameServer::~CGameServer()
{
	if (m_mqConnection)
		releasemq();
}

bool CGameServer::InitGame()
{
	char config[] = "./config.xml";
	if (!ReadCommonConf(config))
	{
		return false;
	}

	return initmq();
}

int CGameServer::OnBYTimeOut(int Timerid)
{
	return 0;
}

int CGameServer::ProcessPacket(SocketHandler * pHandler, NETInputPacket* inputPkg)
{
	unsigned int cmd = inputPkg->GetCmdType();
    log_debug("cmd=%d;", cmd);

	switch (cmd)
	{
		case gameinfo_storage_proto::GAMEINFO_STORAGE_CMD_EVENT_REQ:
			{
				procEvent(pHandler, inputPkg);
				break;
			}
		default:
			break;
	}

	return 0;
}


bool CGameServer::initmq(void)
{
	// 创建MQ链接对象
    m_mqConnection = amqp_new_connection();
    if (!m_mqConnection)
    {
        log_error("create mq conn err");
        return false;
    }

	// 创建TCP socket
    amqp_socket_t* socket = amqp_tcp_socket_new(m_mqConnection);
    if (!socket)
    {
        amqp_destroy_connection(m_mqConnection);
        m_mqConnection = nullptr;
        log_error("create mq TCP socket err");
        return false;
    }

	// 链接到MQ服务器
    int status = amqp_socket_open(socket, m_mqIp.c_str(), m_mqPort);
    if (status)
    {
        amqp_destroy_connection(m_mqConnection);
        m_mqConnection = nullptr;
        log_error("open mq socket err");
        return false;
    }

	// MQ登录认证　用户名、密码、虚拟主机
    amqp_rpc_reply_t x = amqp_login(m_mqConnection, m_mqVhost.c_str(), 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, m_mqUsername.c_str(), m_mqPassword.c_str());
    if (AMQP_RESPONSE_NORMAL != x.reply_type)
    {
        amqp_connection_close(m_mqConnection, AMQP_REPLY_SUCCESS);
        amqp_destroy_connection(m_mqConnection);
        m_mqConnection = nullptr;
        log_error("mq login fail,host:%s,user:%s,pwd:%s", m_mqVhost.c_str(), m_mqUsername.c_str(), m_mqPassword.c_str());
        return false;
    }

	// 打开MQ通道
    amqp_channel_open_ok_t* t = amqp_channel_open(m_mqConnection, m_mqChannel);
    if (!t)
    {
        amqp_connection_close(m_mqConnection, AMQP_REPLY_SUCCESS);
        amqp_destroy_connection(m_mqConnection);
        m_mqConnection = nullptr;
        log_error("open mq channel fail");
        return false;
    }

	// 检查通道打开是否正常
    amqp_rpc_reply_t x1 = amqp_get_rpc_reply(m_mqConnection);
    if (AMQP_RESPONSE_NORMAL != x1.reply_type)
    {
        releasemq();
        log_error("mq open channel amqp_get_rpc_reply fail");
        return false;
    }

    // 声明所有业务交换机并检查
    const char* exchanges[] = {
        m_mqExchange_NonBusiness,		// 非业务事件 交换机
        m_mqExchange_Business,			// 业务事件   交换机
    };
    for (int i = 0; i < 2; ++i)
	{
		// 声明交换机
        amqp_exchange_declare(
			m_mqConnection,
			m_mqChannel,
			amqp_cstring_bytes(exchanges[i]),
			amqp_cstring_bytes(m_mqExchangeType),
			false, true, false, false, amqp_empty_table);
		// 检查声明结果
        amqp_rpc_reply_t reply = amqp_get_rpc_reply(m_mqConnection);
        if (AMQP_RESPONSE_NORMAL != reply.reply_type) {
            releasemq();
            log_error("amqp_exchange_declare fail, exchange: %s", exchanges[i]);
            return false;
        }
    }

    // 声明队列和路由键数组
    const char* queues[] = {
        m_mqQueue_NonBusiness,
        m_mqQueue_Business
    };
    const char* routingKeys[] = {
        m_mqRoutingKey_NonBusiness,
        m_mqRoutingKey_Business
    };
	// 绑定路由键
    for (int i = 0; i < 2; ++i)
	{
		// 声明队列
        amqp_queue_declare(
			m_mqConnection,
			m_mqChannel,
			amqp_cstring_bytes(queues[i]),
			false, true, false, false, amqp_empty_table);
		// 检查队列声明结果
        amqp_rpc_reply_t reply1 = amqp_get_rpc_reply(m_mqConnection);
        if (AMQP_RESPONSE_NORMAL != reply1.reply_type) {
            releasemq();
            log_error("amqp_queue_declare fail, queue: %s", queues[i]);
            return false;
        }
		// 队列绑定到交换机
        amqp_queue_bind(
			m_mqConnection,
			m_mqChannel,
			amqp_cstring_bytes(queues[i]),
			amqp_cstring_bytes(exchanges[i]),
			amqp_cstring_bytes(routingKeys[i]),
			amqp_empty_table);
		// 检查绑定结果
        amqp_rpc_reply_t reply2 = amqp_get_rpc_reply(m_mqConnection);
        if (AMQP_RESPONSE_NORMAL != reply2.reply_type) {
            releasemq();
            log_error("amqp_queue_bind fail, queue: %s, exchange: %s, routingKey: %s", queues[i], exchanges[i], routingKeys[i]);
            return false;
        }
    }

    return true;
}

void CGameServer::releasemq(void)
{
	if (m_mqConnection)
	{
		amqp_channel_close(m_mqConnection, m_mqChannel, AMQP_REPLY_SUCCESS);
		amqp_connection_close(m_mqConnection, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(m_mqConnection);
		m_mqConnection = nullptr;
	}
	return;
}

bool CGameServer::ReadCommonConf(const char* strxml)
{
	CMarkupSTL  markup;
	if (!markup.Load(strxml))
	{
		log_error("Load xml %s failed", strxml);
		return false;
	}

	if (!markup.FindElem("application"))
	{
		log_error("Not Found Elem [application] in xml[%s]", strxml);
		return false;
	}

	if (!markup.IntoElem())
	{
		log_error("IntoElem [application] failed.");
		return false;
	}


	if (!markup.FindElem("mq"))
	{
		log_error("Not Found Elem [mq] in xml[%s]", strxml);
		return false;
	}
	else
	{
		m_mqIp = markup.GetAttrib("ip");
		m_mqPort = atoi(markup.GetAttrib("port").c_str());
		m_mqVhost = markup.GetAttrib("vhost");
		m_mqUsername = markup.GetAttrib("username");
		m_mqPassword = markup.GetAttrib("password");
	}

	if (m_mqIp.empty() || m_mqPort == 0 || m_mqVhost.empty() || m_mqUsername.empty() || m_mqPassword.empty() || m_mqPassword.empty())
		return false;

	log_debug("ReadCommonConf succ!!");

	return true;
}


int CGameServer::procEvent(SocketHandler* pHandler, NETInputPacket* inputPkg)
{
    gameinfo_storage_proto::EventReq req;
    if (!inputPkg->parsePb(&req)) {
        log_error("parse EventReq fail");
        return -1;
    }
    log_debug("req=%s;", req.DebugString().c_str());

    // 非业务类型事件
    const gameinfo_storage_proto::Event& event = req.event_id();
    if(gameinfo_storage_proto::Event_IsValid(event)) // 修正枚举校验
    {
        if(event == gameinfo_storage_proto::Event::INROOM_EVENT)
        {
            // 非业务事件
            PushPbToMQ(m_mqExchange_NonBusiness, m_mqRoutingKey_NonBusiness, req);
        }
        else
        {
            // 业务事件
            PushPbToMQ(m_mqExchange_Business, m_mqRoutingKey_Business, req);
        }
    }
    return 0;
}

int CGameServer::PushPbToMQ(const std::string& exchange, const std::string& routingKey, const google::protobuf::Message& pbMsg)
{
    if (!m_mqConnection) {
        if (!initmq()) {
            log_error("MQ not connected and reconnect failed");
            return -1;
        }
    }
    std::string binData;
    if (!pbMsg.SerializeToString(&binData)) {
        log_error("protobuf serialize failed");
        return -2;
    }
    amqp_basic_properties_t props;
    props._flags = AMQP_BASIC_DELIVERY_MODE_FLAG;
    props.delivery_mode = 2; // 持久化
    amqp_bytes_t message;
    message.bytes = const_cast<char*>(binData.data());
    message.len = binData.size();
    int ret = amqp_basic_publish(
        m_mqConnection,
        m_mqChannel,
        amqp_cstring_bytes(exchange.c_str()),
        amqp_cstring_bytes(routingKey.c_str()),
        0, 0, &props, message);
    if (AMQP_STATUS_OK != ret) {
        log_error("amqp_basic_publish fail, ret:%d, msg len:%zu", ret, binData.size());
        if (AMQP_STATUS_CONNECTION_CLOSED == ret || AMQP_STATUS_SOCKET_ERROR == ret) {
            releasemq();
            if (initmq()) {
                ret = amqp_basic_publish(
                    m_mqConnection,
                    m_mqChannel,
                    amqp_cstring_bytes(exchange.c_str()),
                    amqp_cstring_bytes(routingKey.c_str()),
                    0, 0, &props, message);
                log_alert("mq reconnect OK, publish ret:%d", ret);
            } else {
                log_error("reconnect fail, ret:%d", ret);
                return -3;
            }
        }
    }
    return ret;
}

